package com.kqkj.dyrule.engine.queryService;

import com.kqkj.dyrule.engine.beans.EventBean;
import com.kqkj.dyrule.engine.beans.EventParam;
import com.kqkj.dyrule.engine.util.EventParamComparator;
import org.apache.flink.api.common.state.ListState;

import java.util.List;

public class StateQueryServiceImpl implements QueryService{
    ListState<EventBean> beansState;
    public StateQueryServiceImpl(ListState<EventBean> beansState){
        this.beansState = beansState;

    }

    /**
     * 次序在flink查找
     * @param eventSequenceParam 规则
     * @return 计算的次序
     * @throws Exception
     */
    public int queryEventSequence(List<EventParam> eventSequence,long startTs,long endTs) throws Exception {

        Iterable<EventBean> eventBeans = beansState.get();
        System.out.println("========="+eventSequence);
        int i = 0;
        int count =0;
        for (EventBean eventBean : eventBeans) {
            System.out.println(i);
            if (eventBean.getTimeStamp() >= startTs && eventBean.getTimeStamp() <= endTs
             && EventParamComparator.compare(eventSequence.get(i),eventBean)) {
                i++;
                count++;
                if (count == eventSequence.size()) return count;
            }
        }
        return count;
    }

    /**
     * flink统计数目查找
     * @param beansState
     * @param eventParam
     * @param startTs
     * @param endTs
     * @return
     * @throws Exception
     */
    public long stateQueryEventCount(ListState<EventBean> beansState,EventParam eventParam,long startTs,long endTs) throws Exception {
        long count = 0;
        Iterable<EventBean> eventBeans = beansState.get();
        for (EventBean bean : eventBeans) {
            //判断遍历的事件是否落在规则条件的时间区间内
            if (bean.getTimeStamp() >= startTs &&
                    bean.getTimeStamp() <= endTs){
                boolean compare = EventParamComparator.compare(eventParam, bean);
                if (compare) count++;
            }
        }
        return count;
    }
}
